package com.alicp.jetcache.redisson;

import com.alicp.jetcache.*;
import com.alicp.jetcache.external.AbstractExternalCache;
import org.redisson.Redisson;
import org.redisson.api.RBatch;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.redisson.client.RedisAuthRequiredException;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/***
 * @author hxy
 */
public class RedissonCache<K, V> extends AbstractExternalCache<K, V> {

    private static Logger logger = LoggerFactory.getLogger(RedissonCache.class);

    private RedissonCacheConfig config;

    private RedissonClient redissonClient;

    Function<Object, byte[]> valueEncoder;
    Function<byte[], Object> valueDecoder;

    public RedissonCache(RedissonCacheConfig config) {
        super(config);
        this.config = config;
        this.valueEncoder = config.getValueEncoder();
        this.valueDecoder = config.getValueDecoder();

        Config redissonConfig = new Config();
        SingleServerConfig singleConfig = redissonConfig.useSingleServer();
        singleConfig.setAddress(config.getAdress());
        singleConfig.setDatabase(config.getDatabase());
        singleConfig.setPassword(config.getPassword());
        singleConfig.setConnectionPoolSize(config.getConnectionPoolSize());
        redissonConfig.setCodec(config.getCodec());
        redissonClient = Redisson.create(redissonConfig);

        if (config.isExpireAfterAccess()) {
            throw new CacheConfigException("expireAfterAccess is not supported");
        }
    }


    @Override
    public CacheConfig<K, V> config() {
        return config;
    }

    @Override
    public <T> T unwrap(Class<T> clazz) {
        throw new IllegalArgumentException(clazz.getName());
    }

    protected String buildStrKey(K key) {
        return config.getKeyPrefix() + key.toString();
    }

    @Override
    protected CacheGetResult<V> do_GET(K key) {
        RBucket<OpenCacheValueHolder<V>> rb = redissonClient.getBucket(buildStrKey(key));
        OpenCacheValueHolder<V> holder = rb.get();
        if (holder != null) {
            if (System.currentTimeMillis() >= holder.getExpireTime()) {
                return CacheGetResult.EXPIRED_WITHOUT_MSG;
            }
            CacheValueHolder cacheValueHolder = new CacheValueHolder();
            cacheValueHolder.setValue(holder.getValue());
            cacheValueHolder.setAccessTime(holder.getAccessTime());
            cacheValueHolder.setExpireTime(holder.getExpireTime());
            return new CacheGetResult(CacheResultCode.SUCCESS, null, cacheValueHolder);
        } else {
            return CacheGetResult.NOT_EXISTS_WITHOUT_MSG;
        }
    }

    @Override
    protected MultiGetResult<K, V> do_GET_ALL(Set<? extends K> keys) {
        Map<K, CacheGetResult<V>> resultMap = new HashMap<>();
        keys.forEach(key -> {
            RBucket<OpenCacheValueHolder<V>> rb = redissonClient.getBucket(buildStrKey(key));
            OpenCacheValueHolder<V> holder = rb.get();
            if (holder != null) {
                if (System.currentTimeMillis() >= holder.getExpireTime()) {
                    resultMap.put(key, CacheGetResult.EXPIRED_WITHOUT_MSG);
                } else {
                    CacheValueHolder cacheValueHolder = new CacheValueHolder();
                    cacheValueHolder.setValue(holder.getValue());
                    cacheValueHolder.setAccessTime(holder.getAccessTime());
                    cacheValueHolder.setExpireTime(holder.getExpireTime());
                    CacheGetResult<V> r = new CacheGetResult<V>(CacheResultCode.SUCCESS, null, cacheValueHolder);
                    resultMap.put(key, r);
                }
            } else {
                resultMap.put(key, CacheGetResult.NOT_EXISTS_WITHOUT_MSG);
            }
        });
        return new MultiGetResult<K, V>(CacheResultCode.SUCCESS, null, resultMap);
    }


    @Override
    protected CacheResult do_PUT(K key, V value, long expireAfterWrite, TimeUnit timeUnit) {
        RBucket<OpenCacheValueHolder<V>> rb = redissonClient.getBucket(buildStrKey(key));
        OpenCacheValueHolder<V> holder = new OpenCacheValueHolder(value, timeUnit.toMillis(expireAfterWrite));
        rb.set(holder, expireAfterWrite, TimeUnit.MILLISECONDS);
        return CacheResult.SUCCESS_WITHOUT_MSG;
    }

    @Override
    protected CacheResult do_PUT_ALL(Map<? extends K, ? extends V> map, long expireAfterWrite, TimeUnit timeUnit) {
        RBatch batch = redissonClient.createBatch();
        map.forEach((k, v) -> {
            OpenCacheValueHolder<V> holder = new OpenCacheValueHolder(v, timeUnit.toMillis(expireAfterWrite));
            batch.getBucket(buildStrKey(k)).setAsync(holder, expireAfterWrite, TimeUnit.MILLISECONDS);
        });
        batch.execute();
        return CacheResult.SUCCESS_WITHOUT_MSG;
    }

    @Override
    protected CacheResult do_REMOVE(K key) {
        RBucket<OpenCacheValueHolder<V>> rb = redissonClient.getBucket(buildStrKey(key));
        rb.delete();
        return CacheResult.SUCCESS_WITHOUT_MSG;
    }

    @Override
    protected CacheResult do_REMOVE_ALL(Set<? extends K> keys) {
        RBatch batch = redissonClient.createBatch();
        keys.forEach(key -> {
            batch.getBucket(buildStrKey(key)).deleteAsync();
        });
        batch.execute();
        return CacheResult.SUCCESS_WITHOUT_MSG;
    }

    @Override
    protected CacheResult do_PUT_IF_ABSENT(K key, V value, long expireAfterWrite, TimeUnit timeUnit) {
        RBucket<OpenCacheValueHolder<V>> rb = redissonClient.getBucket(buildStrKey(key));
        OpenCacheValueHolder<V> holder = new OpenCacheValueHolder(value, timeUnit.toMillis(expireAfterWrite));
        boolean success = rb.trySet(holder, expireAfterWrite, TimeUnit.MILLISECONDS);
        if (success) {
            return CacheResult.SUCCESS_WITHOUT_MSG;
        } else {
            return CacheResult.EXISTS_WITHOUT_MSG;
        }

    }

    @Override
    protected boolean needLogStackTrace(Throwable e) {
        if (e instanceof RedisAuthRequiredException) {
            return false;
        }
        return true;
    }
}
